package com.space.user.mq;

import com.space.component.client.DomainEvent;
import org.apache.rocketmq.common.message.MessageConst;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.cloud.stream.function.StreamBridge;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.util.MimeTypeUtils;

/**
 * @author <a href="mailto:414563035@qq.com">Bedroom</a>
 * @date 2021/4/28 3:03 下午
 */
@Service
public class DomainEventBusImpl implements DomainEventBus {
    @Autowired
    StreamBridge streamBridge;

    @Override
    public <T extends DomainEvent> void fire(T event) {
        final Message<T> message = MessageBuilder.withPayload(event)
                .setHeader(MessageConst.PROPERTY_TAGS, event.getTag())
                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
// 发送到动态目的地
//                .setHeader("spring.cloud.stream.sendto.destination", event.getTopic())
                .build();
        streamBridge.send(event.getTopic(), message);
    }

//    @Bean
//    public P a() {
//        return new P();
//    }
//
//    @Bean
//    public Function<Flux<Message<String>>, Message<String>> demo() {
//        return flux -> flux.map(message -> {
//            System.out.println("我是消费者demo" + message.getPayload());
//            final Message<String> m = MessageBuilder.withPayload("bbb").build();
//            return m;
//        }).blockFirst();
//    }
//
//    static class P implements CommandLineRunner {
//        @Autowired
//        DomainEventBus domainEventService;
//
//        @Autowired
//        StreamBridge streamBridge;
//
//        @Override
//        public void run(String... args) throws Exception {
//            System.out.println("--------");
//            CreateUserAccountEvent createUserAccountEvent = CreateUserAccountEvent.builder()
//                    .userId(10)
//                    .build();
//            final Message<String> message = MessageBuilder.withPayload("aaa").build();
////            domainEventService.fire(createUserAccountEvent);
//            streamBridge.send("demo-out-0", message);
//        }
//    }
//
//    public <T> void sendObject(T msg, String tag) throws Exception {
//        Message message = MessageBuilder.withPayload(msg)
//                .setHeader(MessageConst.PROPERTY_TAGS, tag)
//                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON)
//                .build();
//        source.output1().send(message);
//    }
//
//    public <T> void sendTransactionalMsg(T msg, int num) throws Exception {
//        MessageBuilder<T> builder = MessageBuilder.withPayload(msg)
//                .setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON);
//        builder.setHeader("test", String.valueOf(num));
//        builder.setHeader(RocketMQHeaders.TAGS, "binder");
//        Message<T> message = builder.build();
//        source.output2().send(message);
//    }
}
